package org.jdcenter.rmi.impl;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.rmi.AlreadyBoundException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static org.jdcenter.rmi.impl.Glossary.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jdcenter.rmi.RMIWrap;
import org.jdcenter.rmi.abs.*;


public class LocalAsyncServer implements RpcChannel, Server {
    private Log log = LogFactory.getLog(getClass());

    private String server;
    private Object rmiObj = null;
    private RMIWrap rmiWrap = null;
    private HandlerChannelCache channels;
    private ArrayBlockingQueue<Invoker> invokers = new ArrayBlockingQueue<Invoker>(300);
    private ExecutorService executor;
    private Boolean execute = false;

    public LocalAsyncServer(String server, Object obj, LocalServerFactory factory) {
        assert obj != null;

        this.server = server;
        if(RMIWrap.class.isAssignableFrom(obj.getClass())){
        	this.rmiWrap = (RMIWrap)obj;
        	this.rmiObj = null;
        }else{
        	this.rmiObj = obj;
        	this.rmiWrap = null;
        }
        
        this.channels = new HandlerChannelCache(factory.getContext());
        start(factory);
    }

    public void start(LocalServerFactory factory){
    	executor = Executors.newFixedThreadPool(factory.getContext().getProperty(SERVER_CALLBACK_THREAD_COUNT));
    	execute = true;
    	Thread thread = new Thread(new Runnable(){
    		public void run() {
				while(execute){
		    		try {
		    			Invoker invoker = invokers.poll(500,TimeUnit.MILLISECONDS);
		    			if(invoker != null){
		    				executor.execute(invoker);
		    			}
						
					} catch (InterruptedException e) {
						log.error(e);
					}
		    	}
			}
    	});
    	thread.start();
    }
    
    public String toString() {
        return server(server);
    }

    public void run() throws ServerException {
        try {
            UnicastRemoteObject.exportObject(this);
            Naming.bind(server, this);
            log.info(this + " running");
        } catch (RemoteException e) {
            log.error(e);
            throw new ServerException(e);
        } catch (MalformedURLException e) {
        	log.error(e);
            throw new ServerException(e);
        } catch (AlreadyBoundException e) {
        	log.error(e);
            throw new ServerException(e);
        }
    }

    public void close() throws ServerException {
        try {
            Naming.unbind(server);
            UnicastRemoteObject.unexportObject(this, true);
            execute = false;
            executor.shutdownNow();
            log.info(this + " closed");
        } catch (MalformedURLException e) {
        	log.error(e);
            throw new ServerException(e);
        } catch (NotBoundException e) {
        	log.error(e);
            throw new ServerException(e);
        } catch (RemoteException e) {
        	log.error(e);
            throw new ServerException(e);
        }
    }

    public int advise(RpcHandler handler, int cookie) throws RemoteException, HandlerResumeException {
        if (cookie < 0) {
            HandlerChannel channel = new HandlerChannel(handler);
            int ret = channels.put(channel);
            log.info(client() + " advised " + handler(ret));
            return ret;
        }

        int ret = channels.resume(handler, cookie);
        log.info(client(cookie) + " resumed");
        return ret;
    }
    
    public Object lookup(){
    	log.info("client lookuping ...");
    	if(rmiWrap != null){
    		return new InvokeResult(rmiWrap);
    	}else{
    		return new InvokeResult(rmiObj);
    	}
    }

    public void unadvise(int cookie) throws RemoteException {
        channels.remove(cookie);
        log.info(client(cookie) + " unadvised");
    }

    public void tick() throws RemoteException {
    	log.info(client() + " ticked");
    }

    public Object invoke(int cookie,String method,Object[] args)
                throws RemoteException, HandlerException, HandlerInvalidException, HandlerRejectedException {
        
    	Channel channel = null;
        if (cookie >= 0)
            channel = channels.get(cookie);

        log.info(client(cookie) + " ==> " + args);
        invokers.add(new Invoker(method,args,channel));
        return null;
    }
    
    private class Invoker implements Runnable{
    	private String method = null;
    	private Object[] args = null;
    	private Channel channel = null;
    	public Invoker(String method,Object[] args,Channel channel){
    		this.method = method;
    		this.args = args;
    		this.channel = channel;
    	}
		
		public String getMethod() {
			return method;
		}

		public void setMethod(String method) {
			this.method = method;
		}

		public Object[] getArgs() {
			return args;
		}

		public void setArgs(Object[] args) {
			this.args = args;
		}

		public Channel getChannel() {
			return channel;
		}
		public void setChannel(Channel channel) {
			this.channel = channel;
		}
		@Override
		public void run() {
			if(rmiWrap != null){
				try {
					channel.invoke(null,null,rmiWrap.invoke(method,args));
				} catch (ChannelException e) {
					log.error(e);
				} catch (HandlerException e) {
					log.error(e);
				}
			}else{
				try {
					channel.invoke(null,null,getRealMethod(rmiObj,method,args).invoke(rmiObj, args));
				} catch (IllegalArgumentException e) {
					log.error(e);
				} catch (IllegalAccessException e) {
					log.error(e);
				} catch (InvocationTargetException e) {
					log.error(e);
				} catch (ChannelException e) {
					log.error(e);
				} catch (HandlerException e) {
					log.error(e);
				}
			}
		}
		protected Method getRealMethod(Object obj,String methodName,Object[] args){
			Class[] clazzes = new Class[args.length];
			for(int i=0;i<args.length;i++){
				clazzes[i] = args[i].getClass();
			}
			Method[] methods = obj.getClass().getMethods();
			boolean flag1 = true;
			boolean flag2 = true;
			Method method2 = null;
			for(Method method : methods){
				if(method.getName().equals(methodName)){
					Class[] types = method.getParameterTypes();
					if(types.length == clazzes.length){
						for(int i=0;i<types.length;i++){
							if(types[i] == clazzes[i]){
								flag1 = flag1 && true;					
							}
							if(types[i].isAssignableFrom(clazzes[i])){
								flag2 = flag2 && true;
							}
						}
						if(flag1){
							return method;
						}
						if(flag2){
							method2 = method;
						}
					}				
				}
			}
			return method2;
		}
    	
    }
}

